package com.shujia.spark.core

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo13Action {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo13Action")

    val sc = new SparkContext(conf)

    val linesRDD: RDD[String] = sc.textFile("data/students.txt")

    /**
     * 转换算子和操作算子区别
     * 1、转换算子是懒执行，需要一个操作算子触发执行
     * 2、操作算子会触发任务执行，每一个操作算子会触发一个job
     * 3、转换算子会返回一个新的rdd, 操作算子返回值不是rdd
     * 4、每一个操作算子都会触发前面代码执行
     *
     */
    val clazzRDD: RDD[String] = linesRDD.map((line: String) => {
      println("map")
      val split: Array[String] = line.split(",")
      val clazz: String = split(4)
      clazz
    })

    /**
     * 1、foreach:操作算子
     *
     */
    clazzRDD.foreach(println)

    /**
     * 2、count: 统计rdd的行数
     */
    val count: Long = clazzRDD.count()
    println(s"count:$count")

    /**
     * 3、collect:将rdd的数据拉取到内存中构建成一个scala集合
     * 如果rdd的数据量很大会出现内存溢出
     *
     */
    val array: Array[String] = clazzRDD.collect()
    println(array.mkString(","))

    /**
     * 4、reduce:全局集合
     *
     */
    val oneRDD: RDD[Int] = clazzRDD.map(c => 1)
    val reduce: Int = oneRDD.reduce((x: Int, y: Int) => x + y)
    println(s"reduce:$reduce")

    /**
     * 5、sum: 求和
     */
    val sum: Double = oneRDD.sum()
    println(s"sum:$sum")

    /**
     * 6、save: 保存数据到hdfs
     */
    //使用代码删除输出路径， hdfs api
    val fileSystem: FileSystem = FileSystem.get(new Configuration())
    val path = new Path("data/clazz")
    //判断是否存在
    if (fileSystem.exists(path)) {
      //删除目录
      fileSystem.delete(path, true)
    }


    clazzRDD.saveAsTextFile("data/clazz")


    while (true) {}

  }

}
